package com.pw.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.app.BaseAppV2;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.OrderDetail;
import com.pw.gmall.realtime.entities.OrderInfo;
import com.pw.gmall.realtime.entities.OrderWide;
import com.pw.gmall.realtime.entities.PaymentWide;
import com.pw.gmall.realtime.function.DimAsyncFunction;
import com.pw.gmall.realtime.utils.DimUtils;
import com.pw.gmall.realtime.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * @Author: linux_future
 * @desc: 加入缓存, 异步调用
 * @since: 2022/3/21
 **/
public class DwmOrderWideAppV3 extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideAppV3().init(2009, 1, "DwmOrderWideAppV3",
                "DwmOrderWideAppV3", Constant.TOPIC_DWD_ORDER_INFO, Constant.TOPIC_DWD_ORDER_DETAIL);
    }

    @Override
    protected void handler(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> stream) {
        //stream.get(Constant.TOPIC_DWD_ORDER_INFO).print("info:");
        //stream.get(Constant.TOPIC_DWD_ORDER_DETAIL).print("detail:");

        //1. 订单表和订单明细表join
        SingleOutputStreamOperator<OrderWide> factStream = factsJoin(stream);
        //factStream.print();

        //2.事实表和维度表join
        //2.加入异步处理
        SingleOutputStreamOperator<OrderWide> dimAsyncStream = factJoinDimAsync(factStream);
        dimAsyncStream.print();

        //3.写kafka数据
        writeToKafka(dimAsyncStream);


    }

    private void writeToKafka(SingleOutputStreamOperator<OrderWide> stream) {
        stream.map(JSON::toJSONString).addSink(KafkaUtils.getKafkaSink(Constant.TOPIC_DWM_ORDER_WIDE));
    }

    private SingleOutputStreamOperator<OrderWide> factJoinDimAsync(SingleOutputStreamOperator<OrderWide> stream) {
        SingleOutputStreamOperator<OrderWide> resultStream = AsyncDataStream.unorderedWait(stream, new DimAsyncFunction<OrderWide>() {
            @Override
            public void addDim(Jedis jedis, Connection conn, OrderWide orderWide, ResultFuture<OrderWide> resultFuture) throws Exception {
                JSONObject userInfo = DimUtils.readDim(jedis, conn, "dim_user_info", orderWide.getUser_id());
                //user_info, base_province, sku_info, spu_info, base_category3, base_trademark
                //user_info
                orderWide.setUser_gender(userInfo.getString("GENDER"));
                orderWide.calcuUserAge(userInfo.getString("BIRTHDAY"));

                //province
                JSONObject province = DimUtils.readDim(jedis, conn, "dim_base_province", orderWide.getProvince_id());
                orderWide.setProvince_area_code(province.getString("AREA_CODE"));
                orderWide.setProvince_3166_2_code(province.getString("ISO_3166_2"));
                orderWide.setProvince_name(province.getString("NAME"));
                orderWide.setProvince_iso_code(province.getString("ISO_CODE"));

                //sku_info,
                JSONObject sku = DimUtils.readDim(jedis, conn, "dim_sku_info", orderWide.getSku_id());
                orderWide.setSku_name(sku.getString("SKU_NAME"));
                orderWide.setSku_price(sku.getBigDecimal("PRICE"));
                orderWide.setSpu_id(sku.getLong("SPU_ID"));
                orderWide.setTm_id(sku.getLong("TM_ID"));
                orderWide.setCategory3_id(sku.getLong("CATEGORY3_ID"));

                // spu_info,
                JSONObject spu = DimUtils.readDim(jedis, conn, "dim_spu_info", orderWide.getSpu_id());
                orderWide.setSpu_name(spu.getString("SPU_NAME"));

                // base_category3,
                JSONObject category3 = DimUtils.readDim(jedis, conn, "dim_base_category3", orderWide.getCategory3_id());
                orderWide.setCategory3_name(category3.getString("NAME"));

                // base_trademark
                JSONObject trademark = DimUtils.readDim(jedis, conn, "dim_base_trademark", orderWide.getTm_id());
                orderWide.setTm_name(trademark.getString("TM_NAME"));
                resultFuture.complete(Collections.singletonList(orderWide));
            }
        }, 0, TimeUnit.SECONDS);

        return resultStream;
    }


    private SingleOutputStreamOperator<OrderWide> factsJoin(HashMap<String, DataStreamSource<String>> stream) {
         /*
         双流join, 有两种:
          1. 窗口join
          2. interval join
            a: keyBy之后
            b: 只支持事件时间
         */
        KeyedStream<OrderInfo, Long> infoStream = stream.get(Constant.TOPIC_DWD_ORDER_INFO).map(obj -> JSON.parseObject(obj, OrderInfo.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((info, ts) -> info.getCreate_ts())).keyBy(info -> info.getId());

        KeyedStream<OrderDetail, Long> detailStream = stream.get(Constant.TOPIC_DWD_ORDER_DETAIL).map(detail -> JSON.parseObject(detail, OrderDetail.class))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((detail, ts) -> detail.getCreate_ts()))
                .keyBy(OrderDetail::getOrder_id);

        //开始join（window，interval）
        SingleOutputStreamOperator<OrderWide> factsJoin = infoStream.intervalJoin(detailStream).between(Time.seconds(-10), Time.seconds(10))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo info, OrderDetail detail, Context ctx, Collector<OrderWide> out) throws Exception {
                        out.collect(new OrderWide(info, detail));
                    }
                });
        return factsJoin;
    }
}
